﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using A2DFramework.CommandBusService.MSMQCommandBus.Configuration;
using System.Messaging;
using System.Threading;
using A2DFramework.SessionService;
using A2DFramework.CommandBusService.MSMQCommandBus.SessionSupport;

namespace A2DFramework.CommandBusService.MSMQCommandBus
{
    public class CommandHandlerListener
    {
        private List<ICommandHandlerProcessor> handlers = new List<ICommandHandlerProcessor>();
        List<Type> cmdTypes = new List<Type>();
        CommandHandlerProxy proxy = new CommandHandlerProxy();

        public void AddHandler(ICommandHandlerProcessor handler)
        {
            this.handlers.Add(handler);
        }
        public void Start()
        {
            this.handlers.ForEach(t => proxy.AddHandler(t));

            CollectCommandTypes();

            //run thread each
            foreach(var cmdType in cmdTypes)
                ThreadPool.QueueUserWorkItem(new WaitCallback(ProcessQueue), cmdType);
        }

        private void CollectCommandTypes()
        {
            foreach (var handler in this.handlers)
            {
                foreach (var t in handler.GetType().GetInterfaces())
                {
                    if (!t.IsGenericType)
                        continue;

                    foreach (var args in t.GetGenericArguments())
                    {
                        if (typeof(BaseCommand).IsAssignableFrom(args))
                        {
                            cmdTypes.Add(args);
                        }
                    }
                }
            }
        }
        public void ProcessQueue(object state)
        {
            Type cmdType = (Type)state;
            string msmq =MSMQCommandBusConfiguration.GetQueueName(cmdType);
            while (!MessageQueue.Exists(msmq))
            {
                Console.WriteLine(string.Format("msmq not exist. Will retry in 0.5s.[{0}]", msmq));
                System.Threading.Thread.Sleep(500);
            }

            MessageQueue mq = new MessageQueue(msmq);
            MessageQueueTransaction tx = new MessageQueueTransaction();
            mq.Formatter = new XmlMessageFormatter(new Type[] { typeof(TransferWrapper) });
            while (true)
            {
                Message msg = mq.Receive();
                TransferWrapper transfer = (TransferWrapper)msg.Body;
                BaseCommand cmd = transfer.ExtractCommand();

                ISessionContext session = new SessionContext(cmd.SessionID, cmd.SessionItems);
                BaseCommandResult cmdResult = proxy.Handler(cmd, session);

                if (msg.ResponseQueue != null)
                {
                    TransferWrapper msmqCmd = new TransferWrapper();
                    msmqCmd.InjectCommandResult(cmdResult);

                    Message msg_reply = new Message();
                    msg_reply.Body = msmqCmd;
                    msg_reply.CorrelationId = msg.Id;
                    msg_reply.Formatter = new System.Messaging.XmlMessageFormatter(new Type[] { typeof(TransferWrapper) });

                    tx.Begin();
                    msg.ResponseQueue.Send(msg_reply, tx);
                    tx.Commit();
                }
            }
        }
    }
}
